<?php

namespace App\Process;

use App\Model\RecordsModel;
use App\Utility\Cache;
use Carbon\Carbon;
use EasySwoole\Component\Process\AbstractProcess;

class MessageCreate extends AbstractProcess
{
    private $isRun = false;
    public function run($arg){
        //定时200ms检测有没有任务，有的话就while死循环执行
        $this->addTick(200,function (){
            if(!$this->isRun){
                $this->isRun = true;
                go(function (){
                    while (true){
                        try{
                            $cache = new Cache();
                            //只有push集合不存在或者为空集合的时候才能进行task的队列消费
                            if($cache->scard('openid_push') == 0){
                                //从队列里取一个小程序opneid集合的名称(task_push队列为小程序标识的列)
                                $task = $cache->onQueue('task_push')->read();
                                if($task){
                                    $model = new RecordsModel();
                                    //更新推送记录的字段（本次推送的总数）
                                    $model->update([
                                        'id'     => $task['id'],
                                        'number' => $cache->scard($task['applet']),
                                        'start_at' => Carbon::now()
                                        ]);
                                    //根据名称取拿到相应openid集合并赋值给新的集合
                                    $cache->sUnionStore('openid_push',$task['applet']);

                                    //设置推送次数
                                    $cache->delete('push_number');
                                    for($i = 1;$i<=15000;$i++){
                                        $cache->sAdd('push_number',$i); //加入队列
                                    }

                                    unset($model);
                                }else{
                                    break;
                                }
                            }else{
                                break;
                            }
                            unset($cache);
                        }catch (\Throwable $throwable){
                            break;
                        }
                    }
                    $this->isRun = false;
                });
                unset($queue);
            }
        });
    }

    public function onShutDown()
    {
        // TODO: Implement onShutDown() method.
    }

    public function onReceive(string $str, ...$args)
    {
        // TODO: Implement onReceive() method.
    }
}
